{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction\n",
    "\n",
    "A log processing system using Fenic's text extraction and semantic enrichment capabilities to transform unstructured logs into actionable incident response data.\n",
    "\n",
    "This pipeline demonstrates log enrichment through multi-stage processing:\n",
    "\n",
    "- Template-based parsing without regex\n",
    "- Service metadata enrichment via joins\n",
    "- LLM-powered error categorization and remediation\n",
    "- Incident severity assessment with business context\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Session Configuration for Log Enrichment\n",
    "\n",
    "This code sets up a session for semantic log enrichment using the fenic library. \n",
    "\n",
    "It defines the application and language model configuration, specifying the use of the \"gpt-4o-mini\" model with custom rate and token limits. The session is then created and initialized, enabling subsequent semantic operations on log data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import fenic as fc\n",
    "from pydantic import BaseModel, Field\n",
    "\n",
    "# Configure session with semantic capabilities\n",
    "config = fc.SessionConfig(\n",
    "    app_name=\"log_enrichment\",\n",
    "    semantic=fc.SemanticConfig(\n",
    "        language_models={\n",
    "            \"mini\" : fc.OpenAIModelConfig(model_name=\"gpt-4o-mini\", rpm=500,tpm=200_000)\n",
    "        }\n",
    "    )\n",
    ")\n",
    "\n",
    "# Create session\n",
    "session = fc.Session.get_or_create(config)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Raw Log and Service Metadata Preparation\n",
    "\n",
    "This section defines two datasets:\n",
    "\n",
    "**Raw application logs**: A list of log messages from various services, covering different log levels (ERROR, WARN, INFO, CRITICAL) and formats, simulating real world operational events.\n",
    "\n",
    "**Service metadata**: A list of dictionaries providing additional context for each service, such as the responsible team, criticality level, and on-call communication channel.\n",
    "\n",
    "These datasets are used as the foundation for log enrichment and analysis.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Raw application logs with different text formats\n",
    "raw_logs_data = [\n",
    "    \"2024-01-15 14:32:01 [ERROR] payment-api: Connection timeout to db-primary.internal:5432 after 30s retries=3 connection_id=conn_789\",\n",
    "    \"2024-01-15 14:32:15 [WARN] user-service: Rate limit exceeded client_ip=192.168.1.100 requests=1205/min endpoint=/api/v1/users\",\n",
    "    \"2024-01-15 14:33:02 [ERROR] order-processor: Payment validation failed order_id=12345 payment_method=credit_card error_code=INVALID_CVV\",\n",
    "    \"2024-01-15 14:35:22 [INFO] auth-service: User login successful user_id=user_67890 session_id=sess_abc123 ip=10.0.1.55\",\n",
    "    \"2024-01-15 14:36:45 [ERROR] notification-service: Failed to send email to user@example.com smtp_error=Connection_refused retries=2\",\n",
    "    \"2024-01-15 14:37:12 [WARN] inventory-service: Low stock alert product_id=SKU_9876 current_stock=5 threshold=10\",\n",
    "    \"2024-01-15 14:38:33 [ERROR] payment-api: Database connection pool exhausted max_connections=50 active=50 pending=15\",\n",
    "    \"2024-01-15 14:39:01 [CRITICAL] order-processor: Circuit breaker opened for payment-gateway failure_rate=85% threshold=80%\",\n",
    "    \"2024-01-15 14:40:15 [ERROR] user-service: Authentication failed user_id=user_12345 reason=invalid_token attempts=3\",\n",
    "    \"2024-01-15 14:41:22 [WARN] cache-service: Redis connection latency high avg_latency=250ms threshold=100ms\",\n",
    "    \"2024-01-15 14:42:33 [ERROR] file-service: Disk space critical mount=/data/uploads usage=95% available=2GB\",\n",
    "    \"2024-01-15 14:43:44 [INFO] metrics-service: Health check passed response_time=45ms status=healthy\",\n",
    "    \"2024-01-15 14:44:55 [ERROR] search-service: Elasticsearch cluster unhealthy nodes_down=2 total_nodes=5\",\n",
    "    \"2024-01-15 14:45:10 [WARN] api-gateway: Request timeout to upstream service=user-service timeout=10s endpoint=/api/v1/profile\",\n",
    "    \"2024-01-15 14:46:20 [ERROR] backup-service: S3 upload failed file=backup_20240115.tar.gz error=AccessDenied bucket=prod-backups\"\n",
    "]\n",
    "\n",
    "# Service metadata for classical enrichment\n",
    "service_metadata_data = [\n",
    "    {\"service_name\": \"payment-api\", \"team_owner\": \"payments-team\", \"criticality\": \"critical\", \"on_call_channel\": \"#payments-oncall\"},\n",
    "    {\"service_name\": \"user-service\", \"team_owner\": \"identity-team\", \"criticality\": \"high\", \"on_call_channel\": \"#identity-alerts\"},\n",
    "    {\"service_name\": \"order-processor\", \"team_owner\": \"commerce-team\", \"criticality\": \"critical\", \"on_call_channel\": \"#commerce-oncall\"},\n",
    "    {\"service_name\": \"auth-service\", \"team_owner\": \"identity-team\", \"criticality\": \"critical\", \"on_call_channel\": \"#identity-alerts\"},\n",
    "    {\"service_name\": \"notification-service\", \"team_owner\": \"platform-team\", \"criticality\": \"medium\", \"on_call_channel\": \"#platform-alerts\"},\n",
    "    {\"service_name\": \"inventory-service\", \"team_owner\": \"commerce-team\", \"criticality\": \"high\", \"on_call_channel\": \"#commerce-oncall\"},\n",
    "    {\"service_name\": \"cache-service\", \"team_owner\": \"platform-team\", \"criticality\": \"high\", \"on_call_channel\": \"#platform-alerts\"},\n",
    "    {\"service_name\": \"file-service\", \"team_owner\": \"platform-team\", \"criticality\": \"medium\", \"on_call_channel\": \"#platform-alerts\"},\n",
    "    {\"service_name\": \"metrics-service\", \"team_owner\": \"observability-team\", \"criticality\": \"medium\", \"on_call_channel\": \"#observability\"},\n",
    "    {\"service_name\": \"search-service\", \"team_owner\": \"data-team\", \"criticality\": \"high\", \"on_call_channel\": \"#data-oncall\"},\n",
    "    {\"service_name\": \"api-gateway\", \"team_owner\": \"platform-team\", \"criticality\": \"critical\", \"on_call_channel\": \"#platform-alerts\"},\n",
    "    {\"service_name\": \"backup-service\", \"team_owner\": \"platform-team\", \"criticality\": \"medium\", \"on_call_channel\": \"#platform-alerts\"}\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating DataFrames for Logs and Metadata\n",
    "\n",
    "Here we convert the raw log messages and service metadata into DataFrames, making them ready for further processing and enrichment. We also print a summary of the pipeline setup, including the number of log entries and metadata records being processed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create DataFrames\n",
    "logs_df = session.create_dataframe({\"raw_message\": raw_logs_data})\n",
    "metadata_df = session.create_dataframe(service_metadata_data)\n",
    "\n",
    "print(\"🚀 Log Enrichment Pipeline\")\n",
    "print(\"=\" * 70)\n",
    "print(f\"Processing {logs_df.count()} log entries with {metadata_df.count()} service metadata records\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 1: Parsing Unstructured Log Messages\n",
    "\n",
    "This section demonstrates the first stage of the log enrichment pipeline: parsing raw, unstructured log messages into structured fields (timestamp, level, service, and message) using a template based extraction method. \n",
    "\n",
    "The parsed results are filtered to include only valid entries and a sample of the structured output is displayed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    " # Stage 1: Parse unstructured logs using template extraction\n",
    "print(\"🔍 Stage 1: Parsing unstructured log messages...\")\n",
    "log_template = \"${timestamp:none} [${level:none}] ${service:none}: ${message:none}\"\n",
    "\n",
    "parsed_df = logs_df.select(\n",
    "    fc.text.extract(\"raw_message\", log_template).alias(\"parsed\")\n",
    ").select(\n",
    "    fc.col(\"parsed\").get_item(\"timestamp\").alias(\"timestamp\"),\n",
    "    fc.col(\"parsed\").get_item(\"level\").alias(\"level\"),\n",
    "    fc.col(\"parsed\").get_item(\"service\").alias(\"service\"),\n",
    "    fc.col(\"parsed\").get_item(\"message\").alias(\"message\")\n",
    ").filter(\n",
    "    fc.col(\"timestamp\").is_not_null()\n",
    ")\n",
    "\n",
    "print(\"Sample parsed logs:\")\n",
    "parsed_df.select(\"timestamp\", \"level\", \"service\").show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 2: Classical Enrichment with Service Metadata\n",
    "\n",
    "In this stage, the parsed log data is enriched by joining it with service metadata. \n",
    "\n",
    "The service name is aligned between datasets, and additional context, such as team ownership, criticality, and on-call channel, is added to each log entry. \n",
    "\n",
    "A sample of the enriched logs is then displayed, showing how operational context is integrated with the raw log information."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 2: Classical enrichment with service metadata\n",
    "print(\"\\n🔗 Stage 2: Enriching with service metadata...\")\n",
    "# Rename service_name to service using select with alias\n",
    "metadata_df_renamed = metadata_df.select(\n",
    "    metadata_df.service_name.alias(\"service\"),\n",
    "    \"team_owner\",\n",
    "    \"criticality\",\n",
    "    \"on_call_channel\"\n",
    ")\n",
    "enriched_df = parsed_df.join(\n",
    "    metadata_df_renamed,\n",
    "    on=\"service\",\n",
    "    how=\"left\"\n",
    ").select(\n",
    "    \"timestamp\",\n",
    "    \"level\",\n",
    "    \"service\",\n",
    "    \"message\",\n",
    "    \"team_owner\",\n",
    "    \"criticality\",\n",
    "    \"on_call_channel\"\n",
    ")\n",
    "\n",
    "print(\"Sample enriched logs:\")\n",
    "enriched_df.select(\"service\", \"timestamp\", \"team_owner\", \"criticality\").show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 3: Semantic Enrichment with Language Models\n",
    "\n",
    "In this stage, advanced semantic enrichment is applied to the log data using large language models (LLMs). \n",
    "\n",
    "The code extracts detailed error analysis fields, classifies the severity of each incident by considering both the message and service criticality, and generates actionable remediation steps for on-call teams. \n",
    "\n",
    "The final output presents a comprehensive, enriched view of each log entry, combining structured metadata, semantic insights, and recommended actions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 3: Semantic enrichment using LLM operations\n",
    "print(\"\\n🧠 Stage 3: Applying semantic enrichment with LLMs...\")\n",
    "print(\"This may take a few moments as we process logs with language models...\")\n",
    "\n",
    "# Define the Pydantic model for semantic error extraction\n",
    "class ErrorAnalysis(BaseModel):\n",
    "    \"\"\"Pydantic model for semantic error extraction\"\"\"\n",
    "    error_category: str = Field(..., description=\"Main category of the error (e.g., database, network, authentication, resource)\")\n",
    "    affected_component: str = Field(..., description=\"Specific component or resource affected\")\n",
    "    potential_cause: str = Field(..., description=\"Most likely root cause of the issue\")\n",
    "\n",
    "# Semantic extraction for error analysis using Pydantic model\n",
    "final_df = enriched_df.select(\n",
    "    \"timestamp\",\n",
    "    \"level\",\n",
    "    \"service\",\n",
    "    \"message\",\n",
    "    \"team_owner\",\n",
    "    \"criticality\",\n",
    "    \"on_call_channel\",\n",
    "    # Extract error analysis information using Pydantic model\n",
    "    fc.semantic.extract(\"message\", ErrorAnalysis).alias(\"analysis\"),\n",
    "    # Classify incident severity based on message and service criticality\n",
    "    fc.semantic.classify(\n",
    "        fc.text.concat(fc.col(\"message\"), fc.lit(\" (criticality: \"), fc.col(\"criticality\"), fc.lit(\")\")),\n",
    "        [\"low\", \"medium\", \"high\", \"critical\"]\n",
    "    ).alias(\"incident_severity\"),\n",
    "    # Generate remediation steps\n",
    "    fc.semantic.map(\n",
    "        \"Generate 2-3 specific remediation steps that the on-call team should take to resolve this issue: {message} | Service: {service} | Team: {team_owner}\"\n",
    "    ).alias(\"remediation_steps\")\n",
    ")\n",
    "\n",
    "# Create readable final output with extracted fields\n",
    "final_readable = final_df.select(\n",
    "    \"timestamp\",\n",
    "    \"level\",\n",
    "    \"service\",\n",
    "    \"message\",\n",
    "    \"team_owner\",\n",
    "    \"criticality\",\n",
    "    \"on_call_channel\",\n",
    "    final_df.analysis.error_category.alias(\"error_category\"),\n",
    "    final_df.analysis.affected_component.alias(\"affected_component\"),\n",
    "    final_df.analysis.potential_cause.alias(\"potential_cause\"),\n",
    "    \"incident_severity\",\n",
    "    \"remediation_steps\"\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    " # Display results\n",
    "print(\"\\n✅ Pipeline Complete! Final enriched logs:\")\n",
    "print(\"-\" * 70)\n",
    "final_readable.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Analytics: Error Category Distribution\n",
    "\n",
    "This section demonstrates basic analytics on the enriched log data by calculating and displaying the distribution of error categories. \n",
    "\n",
    "It groups the final results by error category and shows the count of occurrences for each, providing insights into the most common types of errors in the dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Analytics examples\n",
    "print(\"\\n📈 Analytics Examples:\")\n",
    "print(\"-\" * 70)\n",
    "\n",
    "# Error category distribution\n",
    "print(\"\\nError Category Distribution:\")\n",
    "final_readable.group_by(\"error_category\").agg(fc.count(\"*\").alias(\"count\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Analytics: Incident Severity by Service Criticality\n",
    "\n",
    "This section analyzes the relationship between service criticality and incident severity.\n",
    "\n",
    "It groups the enriched log data by both criticality and severity levels, displaying the count of incidents for each combination. \n",
    "\n",
    "This helps identify which critical services are experiencing the most severe issues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Severity by service criticality\n",
    "print(\"\\nIncident Severity by Service Criticality:\")\n",
    "final_readable.group_by(\"criticality\", \"incident_severity\").agg(fc.count(\"*\").alias(\"count\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Identifying and Displaying High-Priority Incidents\n",
    "\n",
    "This section filters the enriched log data to highlight incidents classified as \"critical\" or \"high\" severity. \n",
    "\n",
    "It displays key details for each high-priority incident, including the affected service, responsible team, severity, on-call channel, and recommended remediation steps. \n",
    "\n",
    "The total number of urgent incidents is also reported, to quickly identify and respond to the most pressing issues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# High-priority incidents requiring immediate attention\n",
    "print(\"\\nHigh-Priority Incidents (Critical/High severity):\")\n",
    "print(\"-\" * 70)\n",
    "critical_incidents = final_readable.filter(\n",
    "    (final_readable.incident_severity == \"critical\") | (final_readable.incident_severity == \"high\")\n",
    ").select(\n",
    "    \"service\",\n",
    "    \"team_owner\",\n",
    "    \"incident_severity\",\n",
    "    \"on_call_channel\",\n",
    "    \"remediation_steps\"\n",
    ")\n",
    "critical_incidents.show()\n",
    "\n",
    "critical_count = critical_incidents.count()\n",
    "print(f\"\\n🚨 Found {critical_count} high-priority incidents requiring immediate attention\")\n",
    "session.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
